package com.right.netty;

import com.right.factory.PackageFactory;
import com.right.handler.PackageHandler;
import com.right.netty.pipeline.CompositeMessageHandler;
import io.netty.bootstrap.AbstractBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.AdaptiveRecvByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import lombok.extern.slf4j.Slf4j;

import java.net.SocketAddress;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicReference;


/**********************************************
 * @author pillar
 * @date 2018/3/22
 **********************************************/
@Slf4j
public abstract class AbstractConnector<
        C extends ConnectorConfiguration,
        B extends AbstractBootstrap> {

    private PackageFactory mPackageFactory;
    private CompositeMessageHandler messageHandler;
    private final AtomicReference<Channel> channelAtomicReference = new AtomicReference<>();
    private C configuration;
    private SocketAddress socketAddress;
    private EventLoopGroup bossEventLoopGroup;
    private EventLoopGroup workerEventLoopGroup;
    private B bootstrap;

    protected AbstractConnector(C configuration, PackageFactory packageFactory){

        assert (configuration !=null):"Configuration must be provided";
        Objects.requireNonNull(packageFactory, "PackageFactory must be provided");
        this.configuration = configuration;
        this.mPackageFactory = packageFactory;
        messageHandler = new CompositeMessageHandler();
    }

    public AbstractConnector addHandler(PackageHandler... packageHandler){
        messageHandler.addLast(packageHandler);
        return this;
    }

    public void init(){
        log.info("Initializing");
        bossEventLoopGroup = createBossEventLoopGroup();
        workerEventLoopGroup = createWorkerEventLoopGroup();
        bootstrap = createBootstrap();
    }

    protected B getBootstrap() {
        return bootstrap;
    }

    protected void configureBootstrap(B bootstrap) {
        bootstrap.option(ChannelOption.AUTO_READ, true)
                .option(ChannelOption.RCVBUF_ALLOCATOR, AdaptiveRecvByteBufAllocator.DEFAULT)
                .option(ChannelOption.SO_RCVBUF, 512)
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
         .option(ChannelOption.TCP_NODELAY, true);
    }

    public void shutdown() {
        if (workerEventLoopGroup != null) {
            workerEventLoopGroup.shutdownGracefully();
            workerEventLoopGroup = null;
        }
        if (bossEventLoopGroup != null) {
            bossEventLoopGroup.shutdownGracefully();
            bossEventLoopGroup = null;
        }
    }
    protected EventLoopGroup createBossEventLoopGroup() {
        return new NioEventLoopGroup(1);
    }
    protected EventLoopGroup createWorkerEventLoopGroup() {
        return new NioEventLoopGroup(Runtime.getRuntime().availableProcessors());
    }

    /**
     * bootstrap
     * @return
     */
    protected abstract B createBootstrap();

    public SocketAddress getSocketAddress() {
        return socketAddress;
    }

    public void setSocketAddress(SocketAddress socketAddress) {
        this.socketAddress = socketAddress;
    }

    public PackageFactory getMessageFactory() {
        return mPackageFactory;
    }

    protected CompositeMessageHandler getMessageHandler() {
        return messageHandler;
    }

    protected Channel getChannel() {
        return channelAtomicReference.get();
    }

    protected void setChannel(Channel channel) {
        this.channelAtomicReference.set(channel);
    }

    public C getConfiguration() {
        return configuration;
    }

    protected EventLoopGroup getBossEventLoopGroup() {
        return bossEventLoopGroup;
    }
    protected EventLoopGroup getWorkerEventLoopGroup() {
        return workerEventLoopGroup;
    }
}
